package cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB06_订单日清日结看板指标 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        /* 开启flinkSql底层自带的提前窗口触发器，每1分钟触发一次*/
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "10000 ms");


        // 建表，映射业务库中 oms_order 表，用cdc连接器
        // 建表，映射业务库中的“订单主表”,用cdc连接器
        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        " id            INT ,             " +  // 订单id
                        " status        INT ,             " +  // 订单状态
                        " total_amount  decimal(10,0) ,   " +  // 订单原价金额
                        " pay_amount    decimal(10,0) ,   " +  // 应付、实付额
                        " create_time   timestamp(3)  ,   " +  // 订单创建时间
                        " payment_time  timestamp(3)  ,   " +  // 订单支付时间
                        " delivery_time timestamp(3)  ,   " +  // 订单发货时间
                        " receive_time  timestamp(3)  ,   " +  // 订单确认时间
                        " modify_time   timestamp(3)  ,   " +  // 数据更新时间
                        " rt as modify_time ,             " +
                        " watermark for rt as rt - interval '0' second ,   " +
                        " PRIMARY KEY (id) NOT ENFORCED            " +
                        " ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw' ,         " +
                        "     'table-name' = 'oms_order'               " +
                        ")"
        );

        // 过滤出我们需要统计的数据（当日提交的，或当日支付的，或当日发货的，或当日确认的）
        tenv.executeSql(
                " CREATE TEMPORARY VIEW filtered  AS  SELECT                        "+
                        "     *                                                              "+
                        " from order_mysql                                                   "+
                        " where                                                              "+
                        "     TO_DATE(DATE_FORMAT(create_time,'yyyy-MM-dd'))=CURRENT_DATE    "+
                        "  OR TO_DATE(DATE_FORMAT(payment_time,'yyyy-MM-dd'))=CURRENT_DATE   "+
                        "  OR TO_DATE(DATE_FORMAT(delivery_time,'yyyy-MM-dd'))=CURRENT_DATE  "+
                        "  OR TO_DATE(DATE_FORMAT(receive_time,'yyyy-MM-dd'))=CURRENT_DATE   "
        );


        //
        tenv.executeSql(
                " SELECT                                                                                                                                  "+
                        "     proctime() as pt,                                                                                                                   "+
                        "     COUNT(1) FILTER(WHERE TO_DATE(DATE_FORMAT(create_time,'yyyy-MM-dd'))=CURRENT_DATE) as `当日订单总数`,                               "+
                        " 	  SUM(total_amount) FILTER(WHERE TO_DATE(DATE_FORMAT(create_time,'yyyy-MM-dd'))=CURRENT_DATE) as `当日订单总额`,                        "+
                        " 	  SUM(pay_amount) FILTER(WHERE TO_DATE(DATE_FORMAT(create_time,'yyyy-MM-dd'))=CURRENT_DATE) as `当日订单应付总额`,                      "+
                        "                                                                                                                                         "+
                        "     COUNT(1) FILTER(WHERE TO_DATE(DATE_FORMAT(create_time,'yyyy-MM-dd'))=CURRENT_DATE AND status=0) as `当日待支付订单总数`,            "+
                        "     SUM(total_amount) FILTER(WHERE TO_DATE(DATE_FORMAT(create_time,'yyyy-MM-dd'))=CURRENT_DATE AND status=0) as `当日待支付订单总额`,   "+
                        "                                                                                                                                         "+
                        "     COUNT(1) FILTER(WHERE TO_DATE(DATE_FORMAT(payment_time,'yyyy-MM-dd'))=CURRENT_DATE) as `当日已支付订单总数`,                        "+
                        "     SUM(total_amount) FILTER(WHERE TO_DATE(DATE_FORMAT(payment_time,'yyyy-MM-dd'))=CURRENT_DATE ) as `当日已支付订单总额`,              "+
                        "                                                                                                                                         "+
                        "                                                                                                                                         "+
                        "     COUNT(1) FILTER(WHERE TO_DATE(DATE_FORMAT(delivery_time,'yyyy-MM-dd'))=CURRENT_DATE) as `当日发货订单总数`,                         "+
                        "     SUM(total_amount) FILTER(WHERE TO_DATE(DATE_FORMAT(delivery_time,'yyyy-MM-dd'))=CURRENT_DATE ) as `当日发货订单总额`,               "+
                        "                                                                                                                                         "+
                        "     COUNT(1) FILTER(WHERE TO_DATE(DATE_FORMAT(confirm_time,'yyyy-MM-dd'))=CURRENT_DATE) as `当日确认订单总数`,                          "+
                        "     SUM(total_amount) FILTER(WHERE TO_DATE(DATE_FORMAT(confirm_time,'yyyy-MM-dd'))=CURRENT_DATE ) as `当日确认订单总额`                 "+
                        "                                                                                                                                         "+
                        " FROM filtered                                                                                                                           "+
                        " GROUP BY                                                                                                                                "+
                        "     TUMBLE(rt,INTERVAL '24' HOUR)                                                                                                       "
        ).print();
    }
}
